package cn.itcast.dmp.tags

import cn.itcast.dmp.process.Processor
import cn.itcast.dmp.untils.SparkSessionUtils
import org.apache.spark.sql.DataFrame

import scala.collection.mutable
/**
  * 今日ODS表和AREA表生成今日用户标签
  */
object MakeTagsProcessor  extends  Processor{
  override def processData(odsWithAreaDF: DataFrame): DataFrame = {
    import org.apache.spark.sql.functions._

    /**
      * 生成标签数据：广告标识、渠道、关键词、省市、性别、年龄、商圈、App名称和设备
      */
    // 获取SparkSession实例对象
    val spark = SparkSessionUtils.createSparkSession(this.getClass)

    // 针对DataFrame中每个分区数据进行数据标签化操作
    odsWithAreaDF.rdd.mapPartitions{ iter =>
      // 对各个分区数据Row进行数据标签化操作
      iter.map{ row =>
        // 其一：构建集合Map对象，用于存放标签数据
        val tagsMap: mutable.Map[String, Double] = mutable.Map[String, Double]()
        // 1). 广告类型标签（Tags4AdTypeMaker）
        tagsMap ++= Tags4AdTypeMaker.make(row)
        // 2). 渠道标签（Tags4ChannelMaker）
        tagsMap ++= Tags4ChannelMaker.make(row)
        // 3). 关键词标签（Tags4KeyWordsMaker)
        tagsMap ++= Tags4KeyWordsMaker.make(row)
        // 4). 省份城市标签（Tags4RegionMaker）
        tagsMap ++= Tags4RegionMaker.make(row)
        // 5). 性别标签（Tags4GenderMaker）
        tagsMap ++= Tags4GenderMaker.make(row)
        // 6). 年龄标签（Tags4AgeMaker）
        tagsMap ++= Tags4AgeMaker.make(row)
        // 7). 商圈标签（Tags4AreaMaker）
        tagsMap ++= Tags4AreaMaker.make(row)
        // 8). App标签（Tags4AppMaker）
       tagsMap ++= Tags4AppMaker.make(row, appDicBroadcast.value)
        // 9). 设备标签（Tags4DeviceMaker）
        tagsMap ++= Tags4DeviceMaker.make(row, deviceDicBroadcast.value)

        // 其二：每条数据标识符，用字段：uuid
        val mainId: String = row.getAs[String]("uuid")
        // 其三：获取每条数据中所有标识符IDs的值

        // 返回二元组形式
        (mainId, tagsMap.toMap)  // 由于Kudu不支持Map数据类型，所以将Map转换为String类型

      }


    }




    odsWithAreaDF
  }
}
